package org.lds.hotkey.client.holder;

import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.Watch;
import io.etcd.jetcd.watch.WatchEvent;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.lds.hotkey.client.autoconfigure.component.AppContext;
import org.lds.hotkey.client.common.ClientErrorEnum;
import org.lds.hotkey.common.autoconfigure.etcd.EtcdService;
import org.lds.hotkey.common.autoconfigure.spring.SpringUtil;
import org.lds.hotkey.common.constant.KeyConstant;
import org.lds.hotkey.common.enums.ErrorEnum;
import org.lds.hotkey.common.model.Entity;
import org.lds.hotkey.common.model.WorkerModel;
import org.lds.hotkey.common.util.AssertUtil;
import org.lds.hotkey.common.util.EntityModelUtil;
import org.lds.hotkey.common.util.JsonUtil;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class WorkerHolder {
    private static final EtcdService etcdService = SpringUtil.getBean(EtcdService.class);

    private static final ConcurrentHashMap<Long, WorkerModel> WORKER_MAP = new ConcurrentHashMap<>();

    public static void start() {
        AssertUtil.assertTrue(Objects.nonNull(AppContext.appId), ClientErrorEnum.APP_ID_NOT_EXIST);

        init();
        watch();
    }

    /**
     * 初始化 WORKER_MAP
     */
    public static void init() {
        init(KeyConstant.WORKER_DIRECTORY + AppContext.appId);

        if (WORKER_MAP.isEmpty()) {
            init(KeyConstant.WORKER_DIRECTORY + "0");
        }

        log.info("WORKER_SET={}", JsonUtil.toJson(WORKER_MAP));
    }

    private static void init(String keyPath) {
        List<KeyValue> keyValues = etcdService.listByPrefix(keyPath);
        AssertUtil.assertTrue(Objects.nonNull(keyValues), ErrorEnum.ETCD_UNAVAILABLE);

        for (KeyValue keyValue : keyValues) {
            Entity<WorkerModel> entity = EntityModelUtil.getEntity(keyValue, WorkerModel.class);
            if (Objects.nonNull(entity)) {
                WORKER_MAP.put(entity.getId(), entity.getValue());
            }
        }
    }

    public static void watch() {
        watch(KeyConstant.WORKER_DIRECTORY + "0");
        watch(KeyConstant.WORKER_DIRECTORY + AppContext.appId);
    }

    private static void watch(String keyPath) {
        Watch.Watcher watcher = etcdService.watchKey(keyPath, watchResponse -> {
            for (WatchEvent event : watchResponse.getEvents()) {
                Entity<WorkerModel> entity = EntityModelUtil.getEntity(event, WorkerModel.class);
                if (Objects.isNull(entity)) continue;

                if (event.getEventType() == WatchEvent.EventType.PUT) {
                    WORKER_MAP.put(entity.getId(), entity.getValue());
                } else {
                    WORKER_MAP.remove(entity.getId());
                }
            }
            log.info("WORKER_SET={}", JsonUtil.toJson(WORKER_MAP));
        });
        AssertUtil.assertTrue(Objects.nonNull(watcher), ErrorEnum.ETCD_UNAVAILABLE);
    }

    public static WorkerModel getWorkerModel(String key) {
        if (StringUtils.isEmpty(key)) return null;
        List<WorkerModel> workerModels = getWorkerModels();
        if (CollectionUtils.isEmpty(workerModels)) return null;
        int index = key.hashCode() % workerModels.size();
        return workerModels.get(index);
    }

    private static List<WorkerModel> getWorkerModels() {
        return new ArrayList<>(WORKER_MAP.values());
    }

}
